package cn.ac.iie.di.ban.nmpp.handler;

import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
import cn.ac.iie.di.ban.nmpp.tools.MppTools;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

import org.apache.log4j.Logger;
import static cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv.UPDATE_FORWARDMQ;
import static com.datastax.driver.core.DataType.Name.SET;
import com.datastax.driver.core.exceptions.OperationTimedOutException;
import org.json.simple.JSONObject;
import static java.lang.Thread.sleep;

/**
 * Created by jinsheng on 16/9/18.
 *
 */
public class SE2DWWorkerUpsert extends SE2DWWorker {

    private String tableName;
    int insertok = 0;
    private String timestampKey = "";
    private String keySpace;
    long startTime = -1;
//rocketmq Type
    private Map<String, REFieldType> rtn;
    private List<String> updateFields;
    static Map<String, String> ColumnNameType = new HashMap<String, String>();
    private List<ColumnMetadata> columns;
    private Map<String, Type> columnMap = new HashMap<>();
    private ArrayList<Map<String, Object>> updateRecord = new ArrayList<Map<String, Object>>();
    private final ObjectMapper objectMapper = new ObjectMapper();
    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DWWorkerUpsert.class.getName());

    private Cluster cluster;
    private Session session;
//    updateInsert
    private Cluster clusterUI;
    private Session sessionUI;

    private String[] updatePrimaryKeys;
    private PreparedStatement insert_statement = null;
    private PreparedStatement update_statment = null;

    public SE2DWWorkerUpsert(TableSe2DWHandler pTableSe2DWHandler, String pDWType) {
        super(pTableSe2DWHandler, pDWType);
    }

    public List<ColumnMetadata> getColumns() {
        return columns;
    }

    @Override
    public void init() throws Exception {
        tableName = tableSe2DBRule.getTableName();
        if (UPDATE_FORWARDMQ.get(tableName) != null) {
            updateFields = UPDATE_FORWARDMQ.get(tableName).getColumnSet();
            updatePrimaryKeys = UPDATE_FORWARDMQ.get(tableName).getQueryid().split("[:]");
            logger.info(" " + Arrays.toString(updatePrimaryKeys) + " " + updateFields);
        }
        initPStatement();
    }

    private void initConnection() {

        if (cluster != null) {
            return;
        }
        int readTimeOut = 12000;
        Boolean retry = true;
        for (int tryTimes = 0;; tryTimes++) {
            try {
                logger.info(tryTimes + "st times try get available connection to " + tableSe2DBRule.getDbType() + " dw");
                String[] dbmeta = tableSe2DBRule.getConnStr().split("[/]");
                if (dbmeta.length != 3) {
                    logger.error("unknow dbtype " + tableSe2DBRule.getDbType() + " dw unsuccessfully for " + tableSe2DBRule.getConnStr());
                }
                String userName = tableSe2DBRule.getUserName();
                String password = tableSe2DBRule.getPassword();

                QueryOptions options = new QueryOptions();
                options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
                Cluster.Builder builder = Cluster.builder();
                String dataCenter = dbmeta[2];
                keySpace = dbmeta[1];
                tableName = tableSe2DBRule.getTableName();
                String[] Address = dbmeta[0].split("[;]");
                Random ran = new Random();
                int num = ran.nextInt(Address.length);
                String coordinatorAddress = Address[num];
                //创建连接
                //需要用户认证
                if (userName.length() > 0) {
                    builder
                            .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                            .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                            //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                            .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
                    // 不需要用户认证{
                } else {
                    builder
                            .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                            .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                            //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                            .addContactPoint(coordinatorAddress).withQueryOptions(options);
                }
                // .withCredentials(username, passwd).addContactPoint(ip).withQueryOptions(options);
                if (retry == null || retry) {
                    this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
                } else {
                    this.cluster = builder.build();
                }
                break;
            } catch (Exception e) {
                logger.error("cluser create error " + e.getMessage(), e);
                try {
                    cluster.close();
                    sleep(500);
                } catch (Exception ex) {
                    logger.error(ex.getMessage(), ex);
                } finally {
                    cluster = null;
                }
            }
        }

    }

    public void initPStatement() {
        long stinit = System.currentTimeMillis();
        if (cluster == null) {
            for (int tryTimes = 0;; tryTimes++) {
                logger.info(tryTimes + "st times try get valid pstmt to table " + tableSe2DBRule.getTableName() + " on " + tableSe2DBRule.getDbType() + " dw");
                initConnection();
                if (cluster == null) {
                    logger.error("fatal error when getting connection to " + tableSe2DBRule.getDbType() + " dw");
                } else {
                    try {
                        logger.info("create pstmt...");
                        //获取该表的元数据����
                        this.session = this.cluster.connect(keySpace);
                        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
                        logger.info(keySpace + " " + tableName);
                        this.columns = this.cluster.getMetadata().getKeyspace(keySpace).getTable(tableSe2DBRule.getTableName()).getColumns();
                        for (ColumnMetadata name : columns) {
                            columnMap.put(name.getName(), resolveType(name.getType()));
                        }
                        logger.info(columnMap.toString());
                        long etinit = System.currentTimeMillis();
                        logger.info("create pstmt successfully using " + (etinit - stinit) + " ms");
                        if (insert_statement == null) {
                            // 预编译insert语句
                            insert_statement = session.prepare("INSERT INTO " + this.tableName + " JSON :json if not exists ");//if not exists");
                            logger.info("INSERT INTO " + this.tableName + " JSON :json if not exists");
                        }
                        break;
                    } catch (Exception ex) {
                        logger.error("create pstmt to " + tableName + " unsuccessfully for " + ex.getMessage(), ex);
                        close();
                    }
                }
            }
        }
    }

    @Override
    public void loadData0() {
        try {
            if (docRecordSet.size() < 1) {
                logger.warn("no record load to table " + tableSe2DBRule.getTableName() + " in " + dwType);
                return;
            }
            int tryTimes = 0;
            for (tryTimes = 0; tryTimes < 10; tryTimes++) {
                logger.info(tryTimes + "st try to write " + docRecordSet.size() + " records to " + tableSe2DBRule.getTableName() + "...");
                initPStatement();
                logger.info(tryTimes + "st set parameters...");
                String rocketMqColumnName = "";
                String columnName = "";
                ArrayList<Map<String, Object>> DataR = new ArrayList<Map<String, Object>>();
                if (rtn == null) {
                    rtn = RuntimeEnv.ROCKETMQ_DOC_DESC.get(tableName);
                }
                startTime = System.currentTimeMillis();
                for (HashMap docRecord : docRecordSet) {
                    logger.debug("docRecord " + docRecord.toString());
                    Map<String, Object> newSingleData = new HashMap<String, Object>();
                    for (Map.Entry<String, Type> entry : columnMap.entrySet()) {
                        columnName = entry.getKey();
                        rocketMqColumnName = GetColumnName(columnName);
                        if (rocketMqColumnName == null) {
                            continue;
                        }
                        if (docRecord.get(rocketMqColumnName) != null) {
                            Calendar date = Calendar.getInstance();
                            Long insertTime = System.currentTimeMillis() / 1000;
                            Object value = docRecord.get(rocketMqColumnName);
                            if ("it".equals(columnName) || "m_insert_time".equals(rocketMqColumnName)) {
                                newSingleData.put(columnName, insertTime);
                            } else if (rtn.get(rocketMqColumnName) == REFieldType.Struct) {
                                HashMap n = (HashMap) value;
                                String json = JSONObject.toJSONString(n);
                                newSingleData.put(columnName, json);
                            } else if (rtn.get(rocketMqColumnName) == REFieldType.Structs) {
                                Map[] m = (Map[]) value;
                                String jsons = MppTools.maps2json(m);
                                newSingleData.put(columnName, jsons);
                            } else {
                                Object s = resolveType(entry.getValue(), value);
                                newSingleData.put(columnName, value);
                            }
                        } else {

                            logger.error("cannot getName from rocketmq and set columnName null " + columnName + "  : " + rocketMqColumnName);
                            logger.error("testnull " + docRecord.get("g_id") + " " + docRecord.get(rocketMqColumnName));
                            switch (entry.getValue()) {
                                case STRING:
                                    newSingleData.put(columnName, "");
                                    break;
                                case INT:
                                    newSingleData.put(columnName, -1);
                                    break;
                                case BIGINT:
                                    newSingleData.put(columnName, -1l);
                                    break;
                                case LIST:
                                    String[] ss = new String[]{};
                                    newSingleData.put(columnName, ss);
                                    break;
                                default:
                                    logger.error("unknow type " + entry.getValue() + " and column Name is " + columnName);
                            }
                        }
                    }
                    DataR.add(newSingleData);
                }
                logger.debug("data begin loader newmpp and size is " + DataR.size());
                logger.debug(DataR.toString());
                updateRecord.addAll(DataR);
                insertok = 0;
                while (!push_single_json_data(DataR, session, insert_statement, tableName)) {
                    logger.error("insert error  and sleep" + 500 + "ms");
                    close();
                    initPStatement();
                };
                DataR.clear();
                startTime = System.currentTimeMillis();
                ArrayList<Map<String, Object>> DataU = new ArrayList<Map<String, Object>>();
                for (Map oneRecord : updateRecord) {
                    Map<String, Object> upSingleData = new HashMap<String, Object>();
                    for (String colName : updateFields) {
                        Object value = oneRecord.get(colName);
                        if ("ut".equals(colName) || "uut".equals(colName)) {
                            upSingleData.put(colName, System.currentTimeMillis() / 1000);
                        } else if (value != null) {
                            upSingleData.put(colName, value);
                        }
                    }
                    DataU.add(upSingleData);
                }
                logger.debug("data begin loader newmpp and size is " + DataU.size());
                try {
                    while (!updateData(DataU)) {
                        logger.error("update error  and sleep" + 500 + "ms");
                        close();
                        initPStatement();
                    };
                } catch (Exception E) {
                    logger.error(E.getMessage(), E);
                }
                DataU.clear();
                updateRecord.clear();
                docRecordSet.clear();
                break;
            }

        } catch (Exception E) {
            logger.error(E.getMessage(), E);
        }
    }

    private boolean updateData(List<Map<String, Object>> lists) throws InterruptedException {
        List<Map<String, Object>> middle = new ArrayList<>();
        if (lists.size() < 1) {
            return true;
        }
        int update = 0;
        for (int index = 0; index < lists.size(); index += 50) {
            if (index + 50 > lists.size()) {
                middle = lists.subList(index, lists.size());
            } else {
                middle = lists.subList(index, index + 50);
            }
            List<ResultSetFuture> future = new ArrayList<>();
            for (Map<String, Object> map : middle) {
                try {
                    // 预编译update语句
                    Set<String> feilds = map.keySet();
                    if (update_statment == null) {
                        String sql = getUpdatesql(updatePrimaryKeys, feilds);
                        logger.info("update sql is  " + sql);
                        update_statment = session.prepare(sql);
                    }
                    logger.debug(map.toString());
                    //绑定更新数据，并创建BoundStatement对象
                    BoundStatement bs = update_statment.bind();
                    setUpdateValue(map, bs, feilds);
                    // 绑定更新记录的主�?
                    for(String key:updatePrimaryKeys){
                        columnMap.get(key).bind(key, map.get(key), bs);
                    }
                    // 设置记录时间�?
//                    if (this.timestampKey.length() > 0 && map.containsKey(this.timestampKey)) {
//                        bs.setDefaultTimestamp((Long) map.get(this.timestampKey) * 1000000); // second to microsecond
//                    }
                    future.add(session.executeAsync(bs));
                } catch (Exception E) {
                    logger.error(E.getMessage(), E);
                    return false;
                }
            }
            // 处理执行结果
            int trytimes = 0;
            for (int i = 0; i < future.size(); i++) {
                try {
                    ResultSet rs = future.get(i).get();
                    String result_string = rs.toString();
                    if (!result_string.contains("ERROR") && rs.wasApplied()) {
                        update++;
                    } else {
                        logger.info(" update fail " + middle.get(i));
                    }
                } catch (OperationTimedOutException e) {
                    logger.error("Exception " + e.getMessage(), e);
                    i--;
                    trytimes++;
                    sleep(50);
                    if (trytimes >= 5) {
                        trytimes = 0;
                        return false;
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                    return false;
                    // 异常捕捉�?30~234�?
                }
            }
            future.clear();
        }

        Long et = System.currentTimeMillis();
        logger.info("0st load " + lists.size() + "|expect: " + update + " records update to table " + tableName + " and mqName is " + tableSe2DBRule.getMqName() + " successfully using " + (et - startTime) + " ms ");
        return true;
    }

    private String GetColumnName(String getMongodbName) throws IOException, InterruptedException {
        HashMap<String, String> h = RuntimeEnv.TABLE_CR_COLUMNS.get(tableSe2DBRule.getRuleName());

        if (h.get(getMongodbName) != null) {
            return h.get(getMongodbName);
        } else {
            return null;
        }
    }

    private boolean push_single_json_data(List<Map<String, Object>> jsonList, Session uSession, PreparedStatement in_statement, String table) {
        BoundStatement bs;
        List<ResultSetFuture> future = new ArrayList<>();
        try {
            for (Object str : jsonList) {
                // 绑定数据并创建BoundStatement对象
                String json = objectMapper.writeValueAsString(str);
                logger.debug("json" + json);
                bs = in_statement.bind(json);
                // 异步执行查询
                ResultSetFuture result = uSession.executeAsync(bs);
                future.add(result);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return false;
        }
        // 处理语句执行结果
        int tryTimes = 0;
        for (int i = 0; i < future.size(); i++) {
            try {
                // 获取执行结果
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                if (!result_string.contains("ERROR") && rs.wasApplied()) {
                    insertok++;
                    updateRecord.remove(jsonList.get(i));
                }
            } catch (OperationTimedOutException e) {
                logger.error(e.getMessage(), e);
                i--;
                tryTimes++;
                if (tryTimes > 3) {
                    return false;
                }
            } catch (Exception E) {
                logger.error(E.getMessage(), E);
                return false;
            }
        }
        long endTime = System.currentTimeMillis();
        logger.info("0st load " + jsonList.size() + "|expect: " + insertok + " records to table " + table + " and mqName is " + tableSe2DBRule.getMqName() + " successfully using " + (endTime - startTime) + " ms " + updateRecord.size() + " need to update");
        insertok = 0;
        future.clear();
        return true;
    }

    private Type resolveType(DataType rawType) {
        switch (rawType.getName()) {
            case ASCII:
            case VARCHAR:
            case TEXT:
                return Type.STRING;
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Type.INT;
            case BIGINT:
            case VARINT:
                return Type.BIGINT;
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return Type.LIST;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + rawType);

        }
    }

    private enum Type {
        STRING {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else {
                    bs.setString(col, val.toString());
                }
            }
        }, TINYINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
                } else if (val instanceof Number) {
                    bs.setByte(col, ((Number) val).byteValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SMALLINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
                } else if (val instanceof Number) {
                    bs.setShort(col, ((Number) val).shortValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, INT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
                } else if (val instanceof Number) {
                    bs.setInt(col, ((Number) val).intValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BIGINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
                } else if (val instanceof Number) {
                    bs.setLong(col, ((Number) val).longValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, FLOAT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
                } else if (val instanceof Number) {
                    bs.setFloat(col, ((Number) val).floatValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, DOUBLE {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
                } else if (val instanceof Number) {
                    bs.setDouble(col, ((Number) val).doubleValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BLOB {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    try {
                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
                        bs.setBytes(col, ByteBuffer.wrap(bytes));
                    } catch (DecoderException e) {
                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
                    }
                } else if (val instanceof byte[]) {
                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
                } else if (val instanceof ByteBuffer) {
                    bs.setBytes(col, (ByteBuffer) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BOOLEAN {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setBool(col, Boolean.valueOf((String) val));
                } else if (val instanceof Boolean) {
                    bs.setBool(col, (Boolean) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, LIST {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof List) {
                    bs.setList(col, (List<?>) val);
                } else if (val instanceof Collection) {
                    List<?> l = new ArrayList<>((Collection<?>) val);
                    bs.setList(col, l);
                } else {
                    List<Object> l = new ArrayList<>();
                    l.add(val);
                    bs.setList(col, l);
                }
            }
        }, MAP {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Map) {
                    bs.setMap(col, (Map<?, ?>) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SET {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Set) {
                    bs.setSet(col, (Set<?>) val);
                } else if (val instanceof Collection) {
                    Set<?> s = new HashSet<>((Collection<?>) val);
                    bs.setSet(col, s);
                } else {
                    Set<Object> s = new HashSet<>();
                    s.add(val);
                    bs.setSet(col, s);
                }
            }
        };

        public void bind(String col, Object val, BoundStatement bs) {
            throw new IllegalArgumentException("wrong type of " + col);
        }
    }

    /**
     * 提交更新数据到Cassandra
     *
     * @param lists 数据列表
     */
    public void close() {
        try {
            if (session != null) {
                this.session.close();
            }
            if (cluster != null) {
                this.cluster.close();
            }
            sleep(1000);
        } catch (Exception E) {
            logger.error(E.getMessage(), E);
        } finally {
            session = null;
            cluster = null;
            insert_statement = null;
            update_statment = null;
        }
    }

    public void uclose() {
        try {
            if (sessionUI != null) {
                this.sessionUI.close();
            }
            if (clusterUI != null) {
                this.clusterUI.close();
            }
            sleep(1000);
        } catch (Exception E) {
            logger.error(E.getMessage(), E);
        } finally {
            sessionUI = null;
            clusterUI = null;
        }
    }

    /**
     * 映射Cassandra类型到{@linkplain Type}枚举
     *
     * @param rawType Cassandra类型对象
     * @return 返回Cassandra类型对应的Type枚举
     */
    public Session getSession() {
        return this.session;
    }
    //去重程序  调用去重服务

    private Object resolveType(Type tt, Object o) {
        switch (tt) {
            case STRING:
                return o.toString();
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Integer.valueOf(o.toString());
            case BIGINT:
                return Long.valueOf(o.toString());
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return o;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + tt);
        }
    }

    /**
     * 设置更新条件
     *
     * @return
     */
    private String getUpdateStr(Set<String> upFields) {
        StringBuilder sb = new StringBuilder();
        // 遍历�?��更新的各个字段并写入where条件�?
        for (String updateField : upFields) {
            sb.append(updateField)
                    .append("=:")
                    .append(updateField)
                    .append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    /**
     * 设置更新列的�?
     *
     * @param map 从消息队列中读取的数�?
     * @param bs
     */
    private void setUpdateValue(Map map, BoundStatement bs, Set<String> feilds) {
        Set<String> f = new HashSet();
        f.addAll(feilds);
        f.remove("key");
        // 遍历�?��更新的各个列并分别设置更新�?
        for (String updateField : f) {
            Type type = columnMap.get(updateField);
            Object val = map.get(updateField);
            type.bind(updateField, val, bs);
        }
    }

    private String getUpdatesql(String[] querykeys, Set<String> feilds) {
        Set<String> f = new HashSet();
        f.addAll(feilds);
        for(String key:querykeys){
            f.remove(key);
        }
        int length = querykeys.length;
        String sql = "update " + tableName + " set "
                + getUpdateStr(f) + " where ";
        StringBuilder sb = new StringBuilder();
        sb.append(sql);
        int i = 0;
        for (String s : querykeys) {
            sb.append(s);
            i++;
            sb.append("=:").append(s).append(" and ");
        }
        int j = sb.lastIndexOf(" and");
        String ss2 = sb.substring(0, j) + " if exists";
        return ss2;
    }
}
